package com.qdu;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.wal.WALEdit;

import java.io.IOException;
import java.net.URL;
import java.util.List;
import java.util.Optional;

/**
 * 表中插入数据之后，判断是否存在cf1:url列，存在的话，从中提取hostname，并存入另外一个表的中，
 * 并且以hostname作为RowKey，url的内容作为cf1:url的单元格值
 *
 */
public class AnotherUrlHostNameCoprocessor implements RegionObserver, RegionCoprocessor {

    // 用于日志打印，用此对象打印的信息，可用在HBase的RegionServer的日志文件中查看
    private  static  final Log logger = LogFactory.getLog(AnotherUrlHostNameCoprocessor.class);

    // 在当前类实例化时，创建一个连接对象，用于处理触发此协处理器的表之外的别的表,，记得在此类的实例析构掉的时候，关闭
    private Connection connection= null;
    // 要操作的其他表的对象，记得在此类的实例析构掉的时候，关闭
    private Table anotherTable=null;

    @Override
    public Optional<RegionObserver> getRegionObserver() {
        return Optional.of(this);
    }

    @Override
    public void start(CoprocessorEnvironment env) throws IOException {
        connection = ConnectionFactory.createConnection(env.getConfiguration());
        anotherTable = connection.getTable(TableName.valueOf("cls02_index"));
    }

    @Override
    public void stop(CoprocessorEnvironment env) throws IOException {
        anotherTable.close();
        connection.close();
    }

    @Override
    public void postPut(
            ObserverContext<RegionCoprocessorEnvironment> c,
            Put put,
            WALEdit edit,
            Durability durability) throws IOException {
        // 2.1 检查put的内容是否符合要求
        List<Cell> cellList = put.get("cf1".getBytes(),"url".getBytes());
        // 检查cf1:url单元格是否存在
        // 防止操作别的列时，此协处理器继续执行后续的代码
        if(cellList==null || cellList.size()==0){
            return;
        }
        // 2.2 只需要更改put对象中的列对应的单元格即可，不需要执行put等动作命令对应的方法
        // http://www.qdu.edu.cn/news/1.html    ::: url
        // www.qdu.edu.cn           ::: hostname

        // 借助将数据转换为目标类型的对象是否成功，来变相的验证此数据是否格式正确
        try{
            // 从单元格历史列表中取出最新的单元格值
            Cell lastestCell = cellList.get(0);
            String cellValue = new String(CellUtil.cloneValue(lastestCell));
            URL url= new URL(cellValue);

            Put anotherPut = new Put(url.getHost().getBytes());
            anotherPut.addColumn(
                    "cf1".getBytes(),
                    "url".getBytes(),
                    cellValue.getBytes()
            );
            anotherTable.put(anotherPut);
        }
        catch (Exception ex){

            return;
        }
    }
// 在start方法中，创建Connection和Table对象

    // 在值插入表之后，再处理数据存入其他表，需要用到postPUt

}
